package org.ykhl.appserver.services;

import com.google.common.base.Preconditions;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.springframework.web.context.request.async.DeferredResult;
import org.ykhl.appserver.dao.PollRequest;
import org.ykhl.appserver.dao.Request;
import org.ykhl.appserver.exceptions.PollResponseTimeoutException;
import org.ykhl.appserver.exceptions.RequestIdNotFoundException;
import org.ykhl.log.commons.sender.LogSender;
import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.BaseMessageKafkaImpl;
import org.ykhl.mq.commons.ConnectionException;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.pusher.RequestResponse;
import org.ykhl.sdk.appserver.MessageQueueFactoryKafkaImpl;
import org.ykhl.sdk.appserver.MessageQueueKafkaImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ykhl.appserver.exceptions.ValidateException;
import org.ykhl.appserver.model.MessageInfo;


import java.util.HashMap;

import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by zkf on 5/6/16.
 */
public class RequestSendService {
    private static final Logger LOG = LoggerFactory.getLogger(RequestSendService.class);

    private Cache<String, BlockingQueue<BaseMessage>> queueCache; //存放requestId 与queues之间的关系
    private Map<String, MessageQueueKafkaImpl> messageQueues; //存放authId与 queues之间的关系

    private String mqServerAddress = Constants.KAFKA_SERVER_LIST_DEV;
    private String machineID; //该字段标识rest部署在哪个机器上
    private BlockingQueue<BaseMessage> queue = new LinkedBlockingQueue<>();
    private  ValidateService validateService;
    protected LogSender logSender;
    protected Boolean useSsl;
    protected MessageQueueKafkaImpl messageQueue;
    protected RequestResponse requestResponse;
    protected BaseMessage base;
    protected MessageInfo msInfo;
    protected  MessageQueueKafkaImpl authIdQueue;

    public final void setUseSsl(boolean useSsl) {
        this.useSsl = useSsl;
    }

    public void  setMachineID(String machineID){
        this.machineID = machineID;
    }

    public void setMqServerAddress(String mqServerAddress) {
        this.mqServerAddress = mqServerAddress;
    }

    public void setValidateService(ValidateService validateService){
        this.validateService = validateService;
    }

    public final void setLogSender(LogSender logSender) {
        this.logSender = logSender;
    }

    public void init() throws ConnectionException {

        LOG.info("Restful start to init...");

        Preconditions.checkNotNull(mqServerAddress, "mqServerAddress not set");
        Preconditions.checkNotNull(machineID, "machineID not set");
        Preconditions.checkNotNull(useSsl, "useSsl not set");

        LOG.info("mqServerAddress: " + mqServerAddress);
        LOG.info("useSsl: " + useSsl);
        LOG.info("machineID:" + machineID);

        messageQueues = new HashMap<>();
        queueCache = CacheBuilder.newBuilder().recordStats().expireAfterAccess(10, TimeUnit.MINUTES).build();
        LOG.info("init - done");
        logSender.serviceStarted();
    }

    public void destroy() throws ConnectionException {

        ConnectionException ce = null;
        for (String authId : messageQueues.keySet()) {
            try {
                MessageQueueKafkaImpl queue = messageQueues.get(authId);
                queue.stop();
                LOG.info("authId: " + authId + " messageQueue stopped");
            } catch (ConnectionException e) {
                if (ce == null) ce = e;
            }
        }

        if (ce != null) throw ce;
    }


    public BaseMessage getOneResponse(PollRequest pollRequest) {
        LOG.debug("getOneResponse - got a poll request: \n" + pollRequest);

        String reqId = pollRequest.getRequestId();
        BlockingQueue<BaseMessage> queue = queueCache.getIfPresent(reqId);
        if (queue == null)
            throw new RequestIdNotFoundException("no request found by id: " + reqId + ", maybe outdated.");
        BaseMessage base = pollOneMsg(queue, pollRequest.getTimeout(), pollRequest.getTimeUnit());
       // logSender.messageReceiveFromMqHos(base);
        return base;

    }

    public String sendRequest(Request request) throws ConnectionException {
        MessageInfo msInfo =  validateService.isPermissison(request.getAuthId(),request.getFid(),request.getDeviceSign());

        if(msInfo.getCode()==1){
            LOG.debug("sendRequest - got a request: \n" + request);
            request.setMachineID(machineID);
            LOG.debug("sendSingleRequest setMachineID : \n" + request.getMachineID());
            MessageQueueKafkaImpl messageQueue = getMessageQueueByAuthId(request.getAuthId());
            RequestResponse rr = genRequestResponse(request);
            BlockingQueue<BaseMessage> queue = messageQueue.sendRequest(rr);
            queueCache.put(rr.getRequestId(), queue);

            return rr.getRequestId();
        }else{
            LOG.error("appServerId: " + request.getAuthId() + " is not Permission ,Infomation:" + msInfo.getMessage() );
            throw new ValidateException("appServerId: " +  request.getAuthId() + " is not Permission ,Infomation:" + msInfo.getMessage() );
        }
    }

    public BaseMessage sendSingleRequest(Request request) throws ConnectionException {

         msInfo =  validateService.isPermissison(request.getAuthId(),request.getFid(),request.getDeviceSign());
        if(msInfo.getCode()==1){
            LOG.debug("sendSingleRequest - got a request: \n" + request);
            request.setMachineID(machineID);
            LOG.debug("sendSingleRequest getMachineID - new  a request: \n" + request.getMachineID());
             messageQueue = getMessageQueueByAuthId(request.getAuthId());
            requestResponse = genRequestResponse(request);
            queue = messageQueue.sendRequestAsync(requestResponse);
            logSender.messageSentToMqHos(tran2Message(request));//添加日志记录
             base = pollOneMsg(queue, request.getTimeout(), request.getTimeUnit());

//            BaseMessage base = new BaseMessageKafkaImpl();
//            base.setDataType("BS10005");
//            base.setRequestId(rr.getRequestId());
//            base.setMessageType(BaseMessage.MessageType.RESPONSE);
//            base.setAppSecretKey("2hW_H9g3V8ibIqTjlebT_L");
//            base.setDataVersion("5.0");
//            base.setTopic("mq-to-appserver-topic-machineA");
//            base.setFid("BS10005");
//            base.setMessageBody("<ESBEntry>\\n<MessageHeader>\\n<Fid>BS10005</Fid>\\n<SourceSysCode>S01</SourceSysCode>\\n<TargetSysCode>S94</TargetSysCode>\\n<MsgDate>2016-03-03T14:01:33.154746</MsgDate>\\n</MessageHeader>\\n<RetInfo>\\n<RetCode>1</RetCode>\\n<RetCon>查询成功</RetCon>\\n</RetInfo>\\n<MsgInfo>\\n<Msg><![CDATA[<root><head><Fid>BS10005</Fid><title>就诊信息</title><ErrCode>1</ErrCode><ErrMsg>成功</ErrMsg></head><body><ORGAN_CODE>470240332</ORGAN_CODE><PAT_INDEX_NO></PAT_INDEX_NO><VISIT_CARD_NO>9</VISIT_CARD_NO><PAT_NAME>办法吗</PAT_NAME><ID_NUMBER></ID_NUMBER><SEX_CODE>1</SEX_CODE><SEX_NAME>男</SEX_NAME><BIRTH_DATE>1998-02-01 00:00:00</BIRTH_DATE><MOBILE_NO></MOBILE_NO><OUTHOSP_NO>9</OUTHOSP_NO><OUTHOSP_SERIAL_NO>903</OUTHOSP_SERIAL_NO><PAT_TYPE_CODE>1</PAT_TYPE_CODE><PAT_TYPE_NAME></PAT_TYPE_NAME><REGIST_NO></REGIST_NO><REGIST_DATE></REGIST_DATE><VISIT_DATE>2014-02-19 21:07:22</VISIT_DATE><VISIT_START_DATE>2014-02-19 21:06:46</VISIT_START_DATE><VISIT_END_DATE>2014-02-19 21:07:22</VISIT_END_DATE><CHIEF_DESCR></CHIEF_DESCR><CURR_DISEASE_HISTORY></CURR_DISEASE_HISTORY><PAST_DISEASE_HISTORY></PAST_DISEASE_HISTORY><ALLERGY_HISTORY></ALLERGY_HISTORY><RECEPT_TREAT_DR_CODE>21</RECEPT_TREAT_DR_CODE><RECEPT_TREAT_DR_NAME>翁黎荣</RECEPT_TREAT_DR_NAME><VISIT_DEPT_CODE>190</VISIT_DEPT_CODE><VISIT_DEPT_NAME>全科</VISIT_DEPT_NAME><DIAG_CODE>9160</DIAG_CODE><DIAG_NAME>上呼吸道过敏反应</DIAG_NAME></body></root>]]></Msg>\\n<MsgCount>1</MsgCount>\\n<CurrentNum>1</CurrentNum>\\n</MsgInfo>\\n</ESBEntry>\",\n");
//            base.setDeviceSign("LUMIA930-sfagag");
//            base.setAppServerId("2fWVNcnmd7rFLW_2cAVlgz");
//            base.setTimestamp("123456789");
            logSender.messageReceiveFromMqHos(base);
            return base;
        }else{
            LOG.error("appServerId: " + request.getAuthId() + " is not Permission ,Infomation:" + msInfo.getMessage() );
            throw new ValidateException("appServerId: " +  request.getAuthId() + " is not Permission ,Infomation:" + msInfo.getMessage() );
        }
    }


    private BaseMessage pollOneMsg(BlockingQueue<BaseMessage> queue,
                                   long timeout, TimeUnit timeUnit) {

        long start  =System.currentTimeMillis();
        try {
            BaseMessage msg = queue.poll(timeout, timeUnit);
            if (msg == null){
                throw new PollResponseTimeoutException();
            }

            return msg;
        } catch (InterruptedException e) {
            throw new PollResponseTimeoutException();
        }
    }
    private MessageQueueKafkaImpl getMessageQueueByAuthId(String authId)throws ConnectionException {

        authIdQueue = messageQueues.get(authId);
        if (authIdQueue == null) {
             try {
                    authIdQueue = MessageQueueFactoryKafkaImpl.newMachineInstance(authId,machineID);
                    authIdQueue.setUseSsl(false);
                    authIdQueue.setServerAddress(mqServerAddress);
                    authIdQueue.setListener(MessageQueueKafkaImpl.NO_OP_LISTENER);

                    authIdQueue.start();
                    LOG.info("authId: " + authId + " messageQueue started");
                    messageQueues.put(authId, authIdQueue);
                    return authIdQueue;
            } catch (ConnectionException ce) {
                 LOG.error("authId: " + authId + " creat messageQueue error,Exception message:" + ce.getMessage());
                 throw  ce;
            }
        }else{
            return authIdQueue;
        }
    }

    private static RequestResponse genRequestResponse(Request request) {
        return new RequestResponse(request.getFid(), request.getVersion(),
                request.getContent(), request.getDeviceSign(), request.getMachineID());
    }

    /**
     * 收到请求信息日志
     * @param
     */
    public void messageReceivedFromRequest(Request request) {

        logSender.messageReceivedFromRequest(tran2Message(request));
    }

    /**
     * 表示消息已返回到请求方
     * @param baseMessage
     */
    public void messageSentToResponse(BaseMessage baseMessage) {
        logSender.messageSentToResponse(baseMessage);
    }
    /**
     * 将Request对象转换长BaseMessage对象
     * @param request
     * @return
     */
    public BaseMessage tran2Message(Request request){
        BaseMessageKafkaImpl msg = new BaseMessageKafkaImpl();
        msg.setFid(request.getFid());
        msg.setDeviceSign(request.getDeviceSign());
        msg.setAppServerId(request.getAuthId());
        msg.setDataVersion(request.getVersion());
        msg.setRequestId(request.getMachineID());
        msg.setTimestamp("" +System.currentTimeMillis());
        return  msg;
    }
}
